/*
 * lfs_core_impl.c
 *
 *  Created on: Dec 22, 2012
 *      Author: duoln
 */

#include <stdlib.h>
#include <stdio.h>
#include <string.h>
#include <unistd.h>
#include <stdarg.h>
#include <time.h>
#include <sys/time.h>
#include <assert.h>
#include <pthread.h>
#include <dirent.h>
#include <sys/stat.h>
#include <ctype.h>
#include <sys/socket.h>
#include <netinet/in.h>
#include <errno.h>
#include <netinet/tcp.h>
#include <arpa/inet.h>
#include <signal.h>
#include <sys/vfs.h>

#include "lfs.h"
#include "client_remote_master.h"
#include "client_remote_chunkserver.h"
#include "client_meta_cache.h"
#include "logging.h"
#include "iterator_utils.h"
#include "emalloc.h"
#include "queue.h"
#include "stdutils.h"
#include "timer.h"

struct _lfs_
{
    RemoteMaster* master;
    MetaCache* cache;
};

struct _file_stream_
{
    MetaFile* mfile;
    int64_t index;
    int32_t chunk_data_max_size;
    RemoteMaster* master;
    MetaCache* cache;
};
/* core api */

static Logger* log = NULL;

LFS* lfs_init(const char* master_addr, int port)
{
    log = GET_LOGGER("/lfs/client");
    logger(INFO(log), "begin lfs_init");
    LFS* lfs = emalloc(sizeof(LFS));
    lfs->master = remote_master_init(master_addr, port);
    lfs->cache = meta_cache_init();
    logger(INFO(log), "end lfs_init");
    return lfs;
}

void lfs_destory(LFS* lfs)
{
    logger(INFO(log), "begin lfs_destory");
    if (lfs != NULL)
    {
        remote_master_destory(lfs->master);
    }
    logger(INFO(log), "end lfs_destory");
}

Status lfs_fcreate(LFS* lfs, const char* name_space, int32_t min_replica, int32_t max_replica)
{
    logger(INFO(log), "begin lfs_fcreate");
    Status s = LFS_INIT;
    //check param
    if (strlen(name_space) >= FILENAME_LEN)
    {
        logger(WARN(log), "input file name len(%d) >= %d", strlen(name_space), FILENAME_LEN);
        return LFS_INVALID_PARAM;
    }
    s = remote_master_file_create(lfs->master, name_space, min_replica, max_replica);
    //flush cache
    s = meta_cache_flush(lfs->cache, name_space);
    logger(INFO(log), "end lfs_fcreate");
    return s;
}

FileStream* lfs_fopen(LFS* lfs, const char* name_space)
{
    logger(INFO(log), "begin lfs_fopen");
    Status s = LFS_INIT;
    //check param
    if (strlen(name_space) >= FILENAME_LEN)
    {
        logger(WARN(log), "input file name len(%d) >= %d", strlen(name_space), FILENAME_LEN);
        return NULL;
    }
    //get meta from cache
    MetaFile* mfile = NULL;
    if ((mfile = meta_cache_get_meta_file(lfs->cache, name_space)) == NULL)
    {
        if ((s = meta_cache_flush(lfs->cache, name_space)) != LFS_SUCC)
        {
            logger(WARN(log), "CODE:%s", lfs_msg(s));
            return NULL;
        }
        if ((mfile = meta_cache_get_meta_file(lfs->cache, name_space)) == NULL)
        {
            logger(WARN(log), "can not get meta file , namepsace:%s", name_space);
            return NULL;
        }
    }
    //create file stream
    FileStream* fs = emalloc(sizeof(FileStream));
    fs->mfile = mfile;
    fs->index = 0;
    fs->chunk_data_max_size = remote_master_chunk_data_max_size(lfs->master);
    fs->master = lfs->master;
    fs->cache = lfs->cache;
    logger(INFO(log), "end lfs_fopen");
    return fs;
}

Status lfs_fremove(LFS* lfs, const char* name_space)
{
    logger(INFO(log), "begin lfs_fremove");
    Status s = LFS_INIT;
    //check param
    if (strlen(name_space) >= FILENAME_LEN)
    {
        logger(WARN(log), "input file name len(%d) >= %d", strlen(name_space), FILENAME_LEN);
        return LFS_INVALID_PARAM;
    }
    s = remote_master_delete(lfs->master, name_space);
    logger(INFO(log), "end lfs_fremove");
    return s;
}

Status lfs_fwrite(FileStream* file, void* data, int64_t size, int64_t stamp)
{
    logger(INFO(log), "begin lfs_fwrite");
    Status s = LFS_INIT;
    //Cur version only support a chunk to writing
    //there is a big benefit who not need to rollback when one chunk
    if (size > file->chunk_data_max_size)
    {
        logger(WARN(log), "only support one chunk size(%d) to writing but the input size more than the chunk data max size(%d)", size, file->chunk_data_max_size);
        return LFS_INVALID_PARAM;
    }
    //request meta chunk by size
    List* mchunk_list = list_new();
    remote_master_chunk_list_by_size(file->master, size, mchunk_list);
    iterator_list(mchunk_list);
    Queue* queue = queue_create(list_size(mchunk_list));
    Queue* requeue = queue_create(list_size(mchunk_list));
    while (list_has_next(mchunk_list))
    {
        MetaChunk* mchunk = list_next(mchunk_list);
        queue_push(queue, mchunk);
    }
    //write data and the replica info to main chunk
    while (queue_size(queue) > 0)
    {
        //create new thread and send data to chunk server
        MetaChunk* mchunk = queue_pop(queue);
        //to retry queue
        if ((s = remote_chunksever_send_data(NULL, data, size)) != LFS_SUCC)
        {
            queue_push(requeue, mchunk);
        }
    }
    //retry
    while (queue_size(requeue) > 0)
    {
        //create new thread and send data to chunk server
        //MetaChunk* mchunk = queue_pop(requeue);
        //do something
        if ((s = remote_chunksever_send_data(NULL, data, size)) != LFS_SUCC)
        {
            logger(WARN(log), "CODE:%s", lfs_msg(s));
            break;
        }
    }
    //RollBack all operation
    if (s != LFS_SUCC) //Todo
    {
        //not implement
    }
    else
    {
        if ((s = meta_cache_flush(file->cache, file->mfile->namespace)) != LFS_SUCC)
        {
            logger(FATAL(log), "reflush cache fail, CODE:%d", s);
        }
    }
    logger(INFO(log), "end lfs_fwrite");
    return s;
}

Status lfs_fread(FileStream* file, int64_t start, int64_t end, /*out*/List* sfiles)
{
    logger(INFO(log), "begin lfs_fread");
    Status s = LFS_NOT_IMPLEMENT;
    //get meta chunks from cache
    List* mchunks = list_new();
    meta_cache_get_meta_chunks(file->cache, file->mfile->namespace, start, end, mchunks);
    //-----------------------------------
    //-----------------------------------
    //-----------------------------------
    logger(INFO(log), "end lfs_fread");
    return s;
}

Status lfs_fdelete(FileStream* file, int64_t start, int64_t end)
{
    logger(INFO(log), "begin lfs_fdelete");
    Status s = LFS_NOT_IMPLEMENT;
    logger(INFO(log), "end lfs_fdelete");
    return s;
}

Status lfs_fclose(FileStream* file)
{
    logger(INFO(log), "begin lfs_fclose");
    Status s = LFS_INIT;
    logger(INFO(log), "end lfs_fclose");
    return s;
}

/* other method */

Status lfs_get_metafile(LFS* lfs, const char* name_space)
{
    logger(INFO(log), "begin lfs_get_metafile");
    Status s = LFS_NOT_IMPLEMENT;
    logger(INFO(log), "end lfs_get_metafile");
    return s;
}

Status lfs_modify_replica(LFS* lfs, const char* name_space, int32_t min_replica, int32_t max_replica)
{
    logger(INFO(log), "begin lfs_modify_replica");
    Status s = LFS_NOT_IMPLEMENT;
    logger(INFO(log), "end lfs_modify_replica");
    return s;
}

Status lfs_rename_file(LFS* lfs, const char* name_space, const char* new_name_space)
{
    logger(INFO(log), "begin lfs_rename_file");
    Status s = LFS_NOT_IMPLEMENT;
    logger(INFO(log), "end lfs_rename_file");
    return s;
}

inline MetaFile* meta_file_init(const char* namespace, int32_t min_replica, int32_t max_replica)
{
    MetaFile* mfile = emalloc(sizeof(MetaFile));
    strcpy(mfile->namespace, namespace);
    mfile->min_replica = min_replica;
    mfile->max_replica = max_replica;
    mfile->length = 0;
    mfile->mchunks = list_new();
    mfile->create_time = timer_get_micro_second();
    mfile->last_modify_time = mfile->create_time;
    mfile->status = OP_X;
    return mfile;
}

inline char* lfs_msg(int32_t code)
{
    char* msg = NULL;
    switch (code)
    {
    case LFS_INIT:
        msg = LFS_INIT_MSG;
        break;
    case LFS_FAIL:
        msg = LFS_FAIL_MSG;
        break;
    case LFS_SUCC:
        msg = LFS_SUCC_MSG;
        break;
    case LFS_FILE_NOT_EXIST:
        msg = LFS_FILE_NOT_EXIST_MSG;
        break;
    case LFS_FILE_IS_EXIST:
        msg = LFS_FILE_IS_EXIST_MSG;
        break;
    case LFS_PATH_NOT_EXIST:
        msg = LFS_PATH_NOT_EXIST_MSG;
        break;
    case LFS_NOT_IMPLEMENT:
        msg = LFS_NOT_IMPLEMENT_MSG;
        break;
    case LFS_INVALID_PARAM:
        msg = LFS_INVALID_PARAM_MSG;
        break;
    case LFS_LACK_OF_RESOURCE:
        msg = LFS_LACK_OF_RESOURCE_MSG;
        break;
    case LFS_REMOTE_FAIL:
        msg = LFS_REMOTE_FAIL_MSG;
        break;
    case LFS_SERVER_BUSY:
        msg = LFS_SERVER_BUSY_MSG;
        break;
    default:
        msg = LFS_INVALID_ERRORCODE;
        break;
    }
    return msg;
}
